package com.develop.alarm.service;

import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.develop.alarm.bean.AlarmInfo;
import com.develop.common.HttpClientUtil;
import com.develop.common.PropertiesUtils;

import net.sf.json.JSONObject;


  
public class AlarmConsumer {  
	/**
	 * 告警数据保存，从kafka获取最新的告警数据，保存入库
	 * 接收参数如下
	 * topicName:订阅消息主题名称  例如：BJ_ERICSSON_realTimeAlarm
	 * province:省份名称   例如：BJ(北京)
	 * firmName:厂家名称    例如：ERICSSON(爱立信)
	* <p>Description: </p> 
	* @author xxj
	* @date 2019年7月9日
	 */
	public static void saveAlarm(String topicName,String province,String firmName){
		Properties props = new Properties();  
        props.put("bootstrap.servers", PropertiesUtils.getPropertiesByKey("kafka.consumer.bootstrap.servers", "application.properties"));
        props.put("group.id", "test");  
        props.put("enable.auto.commit", "true");  
        props.put("auto.commit.interval.ms", "1000");  
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topicName));
        LinkedList<AlarmInfo> alarmList=new LinkedList<AlarmInfo>();
        //告警批量处理条数
        int batchnum=Integer.parseInt(PropertiesUtils.getPropertiesByKey("batchnum", "application.properties"));
        String msg_url=PropertiesUtils.getPropertiesByKey("msg_url", "application.properties");
        String file_url=PropertiesUtils.getPropertiesByKey("file_url", "application.properties");
        Map<String,String> params=new HashMap<String,String>();
        params.put("cityCode", province);
        params.put("firmCode", firmName);
        params.put("reqId", String.valueOf(System.currentTimeMillis()));
        int connectTimeout=3000;
        int readTimeout=3000; 
        String charset="utf-8"; 
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000); 
            
            /**
             * 实时订阅告警消息处理
             * 1.如果接收的实时告警消息大于等于100条，每次处理100条
             * 2.如果接收的实时告警消息小于100条，全部处理
             * 
             */
            int maxAlarmSeq=AlarmService.getMaxAlarmSeqByProvinceAndFirmName(province, firmName);
            if(records.count()>=batchnum){
            	for (ConsumerRecord<String, String> record : records){
                	JSONObject jsonobject = JSONObject.fromObject(record.value());
                	AlarmInfo alarm=(AlarmInfo)JSONObject.toBean(jsonobject,AlarmInfo.class);
                		alarmList.add(alarm);
                		if(alarmList.size()==batchnum){
                			int begin=alarmList.get(0).getAlarmSeq();
                			int end=alarmList.get(batchnum-1).getAlarmSeq();
                			AlarmService.batchInsertAlarmInfo(alarmList,province, firmName);
                			if(maxAlarmSeq-begin==1&&end-begin==batchnum-1){
                				System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
                				System.out.println("开始告警序号："+begin+"结束告警序号："+end);
                			}else{
                				System.out.println("=============该批次告警数据不连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
                				System.out.println("开始告警序号："+begin+"结束告警序号："+end);
                				
                				//Todo 告知接收程序不连续告警开始序号，告警程序通知告警Socket服务端从不连续告警序号开始重新发送
                				
                				/**
                				 * 消息方式同步告警请求
								 *	http://localhost:8081/ReqSyncAlarm/msg
								 *	参数：
								 *	reqId：操作序号
								 *	cityCode：城市编码
								 *	firmCode：厂家编码
								 *	alarmSeq：起始告警消息序号
                				 */
                				/*params.put("alarmSeq", String.valueOf(begin));
                				String result=HttpClientUtil.get(msg_url, params, connectTimeout, readTimeout, charset);
                				JSONObject jsonResult = JSONObject.fromObject(result);
                				int failCount=0;
                				int state=jsonResult.getInt("state");
                				if(state==1){
                					failCount++;
                				}
                				if(failCount>=3){
                					failCount=0;
                					AlarmService.batchInsertAlarmInfo(alarmList,province, firmName);
                				}*/
                			}
                			alarmList.clear();
                		}
                	}
            }else if(records.count()>0&&records.count()<batchnum){
            	for (ConsumerRecord<String, String> record : records){
		            	JSONObject jsonobject = JSONObject.fromObject(record.value());
		            	AlarmInfo alarm=(AlarmInfo)JSONObject.toBean(jsonobject,AlarmInfo.class);
	            		alarmList.add(alarm);
            			int begin=alarmList.get(0).getAlarmSeq();
            			int end=alarmList.get(batchnum-1).getAlarmSeq();
            			AlarmService.batchInsertAlarmInfo(alarmList,province, firmName);
            			if(maxAlarmSeq-begin==1&&end-begin==batchnum-1){
            				System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
            				System.out.println("开始告警序号："+begin+"结束告警序号："+end);
            			}else{
            				System.out.println("=============该批次告警数据连续======当前系统时间："+new Date()+"告警发生时间："+alarmList.get(0).getEventTime());
            				System.out.println("开始告警序号："+begin+"结束告警序号："+end);
            				//Todo 告知接收程序不连续告警开始序号，告警程序通知告警Socket服务端从不连续告警序号开始重新发送
            				
            			}
            			alarmList.clear();
            	}  
            }
            
        }
	}
    public static void main(String[] args) {  
        String topicName=args[0];
        String province=args[1];
        String firmName=args[2];
        saveAlarm(topicName, province, firmName);
    }
}
